package com.majun.rmq.customer;


import com.majun.service.impl.RocketMqServiceImpl;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

@Component
public class RocketMqConsumer {


    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMqServiceImpl.class);
    /**
     * 发送同一类消息的设置为同一个group，保证唯一,默认不需要设置，rocketmq会使用ip@pid(pid代表jvm名字)作为唯一标示
     */
    @Value("${rocketmq.consumer.groupName}")
    private String consumerGroupName;

    @Value("${rocketmq.consumer.namesrvAddr}")
    private String consumerNamesrvAddr;

    @Value("${topic}")
    private String topic;

    /**
     * 重试次数 暂时不用
     */
    @Value("${reConsumerTimes}")
    private int reConsumerTimes;


    @Bean
    public DefaultMQPushConsumer defaultMQPushConsumer() throws MQClientException {

        // Instantiate with specified consumer group name.
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroupName);
        // Specify name server addresses.
        defaultMQPushConsumer.setNamesrvAddr(consumerNamesrvAddr);
        // Subscribe one more more topics to consume.
        defaultMQPushConsumer.subscribe(topic, "*");
        // Register callback to execute on arrival of messages fetched from brokers.
        defaultMQPushConsumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
            LOGGER.info("%s Receive New Messages:{}", Thread.currentThread().getName(), list);

            try {
                if (list != null && list.size() > 0) {
                    for (MessageExt messageExt : list) {
                        LOGGER.info("messageExt:{}", list.toString());//输出消息内容
                        String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);//读取远程配置的编码为utf-8
                        String tags = messageExt.getTags();
                        LOGGER.info("===>消费响应: msgId:{}, tags :{}, body:{}", messageExt.getMsgId(),tags, messageBody);//输出消息内容


                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
            } catch (Exception e) {
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
            }

        });

        //Launch the consumer instance.
        defaultMQPushConsumer.start();
        LOGGER.info("rocketMq Consumer start success; namesrvAddr:{},groupName:{}", consumerNamesrvAddr, consumerGroupName);
        return defaultMQPushConsumer;
    }
}
